/*
 * REFThreadPool.cpp
 *
 *  Created on: Apr 22, 2013
 *      Author: Lucifer
 */

#include "../pch.h"
#include "REFThread.h"
#include "REFThreadPool.h"

namespace BFX
{

//////////////////////////////////////////////////////////////////////////////

#define NEW_THREAD_THRESHOLD            7	// Number of requests outstanding before we start a new thread

/**
 * Creates a new instance with the given initial parameters.
 * @param[in] nCoreThreads The number of threads to keep in the pool, even if they are idle.
 * @param[in] nMaxThreads The maximum number of threads to allow in the pool.
 * @param[in] nIdleTimeoutInSecounds The expiration value, in seconds. When the number of threads is greater than the core, this
 *		is the maximum time that excess idle threads will wait for new tasks before terminating.
 */
REFThreadPool::REFThreadPool(int nCoreThreads, int nMaxThreads, int nIdleTimeoutInSecounds)
{
	// validate initial arguments
	BFX_ENSURE(nCoreThreads > 0);
	BFX_ENSURE(nMaxThreads > 0);
	BFX_ENSURE(nCoreThreads <= nMaxThreads);
	BFX_ENSURE(nIdleTimeoutInSecounds > 0);

	m_nMinLimitTotalWorkerThreads	= nCoreThreads;
	m_nMaxLimitTotalWorkerThreads	= nMaxThreads;
	m_nIdleWokerTimeout				= (nIdleTimeoutInSecounds * 1000);

	m_nNumWorkerThreads		= 0;
	m_nNumIdleWorkerThreads	= 0;
	m_pWorkRequestNotification	= NULL;
	m_lInitialized	= 0;
	m_bShuttingDown	= false;
}

/**
 * Destroys the instance of thread pool.
 * Before the pool destroyed, It will force to stop all worker threads even they are still alive.
 */
REFThreadPool::~REFThreadPool()
{
	if (!IsInitialized() || m_bShuttingDown)
		return;

	if (m_workRequests.GetSize() > 0)
	{
		// clear work requests forcedly.
		MutexHolder holder (&m_workerCriticalSection);
		m_workRequests.Clear();
	}

	Shutdown();
}

/**
 * Starts thread pool.
 */
void REFThreadPool::Start()
{
	if (IsInitialized())
		return;

	m_pWorkRequestNotification = new REFThreadEvent(false, false);

	// NOTE: No necessary to creates work threads at this point.

	Interlocked::Exchange(&m_lInitialized, -1);
}

/**
 * Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
 */
void REFThreadPool::Shutdown()
{
	if (!IsInitialized() || m_bShuttingDown)
		return;

	m_bShuttingDown = true;
	m_pWorkRequestNotification->Set();
	while (m_nNumWorkerThreads > 0)
	{
		REFThread::Sleep(10);
	}
}

/**
 * Queues a method for execution. The method executes when a thread pool thread becomes available.
 * @param[in] pCallback The pointer of BFX::REFCallback to be executed.
 * @param[in] pContext An pointer containing data to be used by the method.
 * @return True if successful; false if the threadpool has not been started or already shutdown.
 */
bool REFThreadPool::QueueRequest(REFThreadCallback* pCallback)
{
	BFX_REQUIRE(pCallback != NULL);
	if (!IsInitialized() || m_bShuttingDown)
		return false;

	// NOTE: should we check the queue size first, for overflow???
	{
		MutexHolder holder(&m_workerCriticalSection);
		EnqueueWorkRequest(pCallback);

		if (NewWorkerRequired())
		{
			BFX_DBGTRACE("CreateWorker");
			CreateWorkerThread();
		}
	}

	// Between the above enqueue and here, it's possible an active worker thread dequeued the workitem and reset the event,
	// in which case setting the event here causes extra work with no benefit. Ensure there's actually work in the queue before setting.
	if (m_workRequests.GetSize() > 0)
		m_pWorkRequestNotification->Set();

	return true;
}

/**
 * Queues a method for execution. The method executes when a thread pool thread becomes available.
 * @param[in] fnCallback The functor object of method to be executed.
 * @return True if successful; false if the threadpool has not been started or already shutdown.
 */
bool REFThreadPool::QueueRequest(ThreadCallback pfnCallback)
{
	// BFX_REQUIRE1(pfnCallback != NULL, "Null pointer callback disallowed.");

	return QueueRequest(new REFThreadCallback(pfnCallback));
}

// Returns the value indicates whether the pool requires more workers.
bool REFThreadPool::NewWorkerRequired() const
{
	// we only want to grow the worker thread pool if there are less than n threads, where n= no. of processors
	// and more requests than the number of idle threads and number of worker threads not at max ceiling
	return (m_nNumWorkerThreads < m_nMaxLimitTotalWorkerThreads && m_nNumIdleWorkerThreads < m_workRequests.GetSize());
}

// Creates a new worker thread.
void REFThreadPool::CreateWorkerThread()
{
	// creates a new pool worker.
	REF<REFThread> pThread = new REFThread(BFX::Bind(&REFThreadPool::WorkerThreadProc, this));
	pThread->m_bIsPoolThread = true;

	// start the worker.
	pThread->Start();
	pThread->AddRef();	// pool workers want to release themselves when terminated.

	MutexHolder hold(&m_workerCriticalSection);
	m_nNumWorkerThreads ++;
	m_nNumIdleWorkerThreads ++;

	BFX_DBGTRACE("Worker thread created (NumWorkerThreads=%d)\n");
}

// Enqueue a work request.
void REFThreadPool::EnqueueWorkRequest(REFThreadCallback* pWorkRequest)
{
	BFX_DBGTRACE("Enqueue work request");
	m_workRequests.AddLast(pWorkRequest);
}

// Dequeue a work request.
REFThreadCallback* REFThreadPool::DequeueWorkRequest()
{
	REFThreadCallback* pCallback = NULL;

	if (m_workRequests.GetSize() > 0)
	{
		pCallback = m_workRequests.GetFirst();
		pCallback->Autorelease();
		m_workRequests.RemoveFirst();

		BFX_DBGTRACE("Dequeue work request");
	}
	if (m_workRequests.GetSize() == 0)
		m_pWorkRequestNotification->Reset();

	return pCallback;
}

// Execute a work request.
void REFThreadPool::ExecuteWorkRequest(REFThreadCallback* pWorkRequest)
{
	if (pWorkRequest != NULL)
	{
		BFX_DBGTRACE("Starting work request");
		try
		{
			// I hate C++ exceptions.
			// not all system provided the Zero-Cost model exceptions.
			pWorkRequest->Run();
		}
		catch (const Exception & e)
		{
			BFX_DBGTRACE("workrequest throws a %s: %s.", e.GetExceptionName(), e.GetMessage());
		}

		// clear remains for each work request.
		REFAutoreleasePool::Drain();
	}
}

// Thread entry.
void __cdecl REFThreadPool::WorkerThreadProc(REFThreadPool* pThreadPool)
{
	BFX_ASSERT(pThreadPool);
	REFThread* pThread = REFThread::GetCurrentThread();
	BFX_ASSERT(pThread && pThread->m_bIsPoolThread);

	pThreadPool->ThreadProc();
}

#ifndef	_WIN32
#ifdef	__APPLE_CC__
#include <mach/clock.h>
#include <mach/mach.h>

INT32 GetTickCount()
{
	clock_serv_t cs;
	mach_timespec_t mts;
	host_get_clock_service(mach_host_self(), SYSTEM_CLOCK, &cs);
	clock_get_time(cs, &mts);
	mach_port_deallocate(mach_task_self(), cs);
	return (mts.tv_sec * 1000 + mts.tv_nsec / (1000 * 1000));
}

#else	//	__APPLE_CC__
#include <time.h>

INT32 GetTickCount()
{
	struct timespec ts;
	clock_gettime(CLOCK_MONOTONIC, &ts);
	return (ts.tv_sec * 1000 + ts.tv_nsec / (1000 * 1000));
}
#endif	//	__APPLE_CC__
#endif	//	_WIN32

void REFThreadPool::ThreadProc()
{
#define NOWORK_TIMEOUT (10*1000)	// milliseconds

	REFThread* pThread = REFThread::GetCurrentThread();
	BFX_ASSERT(pThread != NULL);

	UINT32 nSleepTime = (UINT32)m_nIdleWokerTimeout;
	UINT32 nLastThreadDequeueTime = GetTickCount();

	for (;;)
	{
		bool bShouldTerminate = false;

		// TODO: should uses SafeWait to prevent exceptions.
		if (!m_pWorkRequestNotification->Wait(nSleepTime))
		{
			// the thread terminates if there are > m_nMinLimitTotalWorkerThreads threads and the queue is small
			if (m_nNumWorkerThreads > m_nMinLimitTotalWorkerThreads)
			{
				// NOTE: we have no necessary to lock before operation.
				int nThreshold = NEW_THREAD_THRESHOLD * (m_nNumWorkerThreads - 1);
				if ((nThreshold >= m_workRequests.GetSize()) && (nSleepTime >= 2 * (UINT32)m_nIdleWokerTimeout))
				{
					BFX_DBGTRACE("ThreadProc: retire idle worker");
					bShouldTerminate = true;
				}
				else
				{
					nSleepTime <<= 1;
					nSleepTime += 1000;	// adds 1 second to prevent wraparound to 0
				}
			}
		}
		else
		{
			// woke up because of a new work request arrival
			REF<REFThreadCallback> pWorkRequest = NULL;
			if (m_workRequests.GetSize() > 0)
			{
				MutexHolder holder(&m_workerCriticalSection);
				pWorkRequest = DequeueWorkRequest();	// the dequeue operation also resets the m_pWorkRequestNotification event

				if (pWorkRequest != NULL)
					m_nNumIdleWorkerThreads --;		// we found work, decrease the number of idle threads
			}
			
			if (!pWorkRequest)
			{
				// we woke up because of a work arrival, but there was no work. this implies too many idle threads and not enough work.
				if ((GetTickCount() - nLastThreadDequeueTime >= NOWORK_TIMEOUT) &&
					m_nNumWorkerThreads > m_nMinLimitTotalWorkerThreads)
				{
					BFX_DBGTRACE("ThreadProc: retire idle worker");
					bShouldTerminate = true;
				}
			}

			// both pWorkRequest and bShouldTerminate cannot be true at the same time
			BFX_ASSERT(!(pWorkRequest != NULL && bShouldTerminate));

			while (pWorkRequest != NULL)
			{
				ExecuteWorkRequest(pWorkRequest);
				nLastThreadDequeueTime = GetTickCount();

				MutexHolder holder(&m_workerCriticalSection);
				pWorkRequest = DequeueWorkRequest();	// the dequeue operation resets the WorkRequestNotification event
				if (pWorkRequest == NULL)
					m_nNumIdleWorkerThreads ++;			// no more work, increase the number of idle threads

				// NOTE: should we reset tls for next work request???
			}
		}

		if (bShouldTerminate || m_bShuttingDown)
		{
			MutexHolder hold(&m_workerCriticalSection);
			BFX_DBGTRACE("Worker thread terminated");

			m_nNumWorkerThreads --;
			m_nNumIdleWorkerThreads --;

			break;	// out of for loop
		}
	}
}

} /* namespace BFX */
